package com.demo.listener;

import com.alibaba.fastjson2.JSON;
import com.demo.MqConstant;
import com.demo.model.dto.OrderMessage;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author shenguangyang
 */
@Component
public class Demo1KafkaListener {
    private static final Logger log = LoggerFactory.getLogger(Demo1KafkaListener.class);
    private final AtomicInteger count = new AtomicInteger(0);

//    @KafkaListener(id = MqConstant.DEMO1_CONSUME_GROUP,
//            groupId = MqConstant.DEMO1_CONSUME_GROUP,
//            topics = MqConstant.DEMO1_TOPIC)
//    public void listen(String message) {
//        Demo1Message demo1Message = JSON.parseObject(message, Demo1Message.class);
//        log.info("input value: {}, count: {}" , demo1Message, count.addAndGet(1));
//    }

    @KafkaListener(id = MqConstant.DEMO1_CONSUME_GROUP,
            groupId = MqConstant.DEMO1_CONSUME_GROUP,
            topics = MqConstant.DEMO1_TOPIC)
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack) {
        OrderMessage orderMessage = JSON.parseObject(record.value(), OrderMessage.class);
        log.info("input value: {}, count: {}" , orderMessage, count.addAndGet(1));
        ack.acknowledge();
    }
}
